package com.example.activemq.consumer;

import com.alibaba.fastjson.JSON;
import com.example.activemq.pojo.MessagePojo;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.activemq.command.ActiveMQMessage;
import org.apache.activemq.util.Callback;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

import javax.jms.JMSException;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.TextMessage;


/**
 * @author LiYuhang
 * @version 0.1
 * @application
 * @Date 2021/4/2 22:50
 * @Description TODO
 */
@Component
@Slf4j
public class QueueConsumerListener {

    @Value("${spring.application.name}")
    private String appName;

    @Value("${server.port:#{80}}")
    private String serverPort;

    /**
     * @param messagePojo
     * @return void
     * @Description: queue 确保消息只被消费一次，未消费数据主动推给监听者
     * @Author: LiYuhang
     * @Date 2021/4/3 00:50
     */
    @JmsListener(destination = "${spring.activemq.queue-name}")
    public void receive(MessagePojo messagePojo) {
        log.info("queue 收到消息 messagePojo:{}",JSON.toJSONString(messagePojo));
    }


    /**
     * queue-test普通队列：消费者1
     */
    @JmsListener(destination = "queue-test", containerFactory = "queueListener")
    public void receiveQueueTest1(ActiveMQMessage message,Session session) throws JMSException {
        if(message instanceof  ObjectMessage){
            ObjectMessage message1 = (ObjectMessage) message;
            MessagePojo pojo = (MessagePojo) message1.getObject();
            log.info(" {}-{} >>>> receiveQueueTest -1 接收到 msgBody:{} message:{}",appName,serverPort,pojo,message);
            sleep(1000);
            // 手动确认
            message.acknowledge();
        }
    }

    /**
     * queue-test普通队列：消费者2
     */
    @JmsListener(destination = "queue-test", containerFactory = "queueListener")
    public void receiveQueueTest2(ActiveMQMessage message,Session session) throws JMSException {
        //TextMessage、MapMessage、BytesMessage、StreamMessage和ObjectMessage
        if(message instanceof  ObjectMessage){
            ObjectMessage message1 = (ObjectMessage) message;
            MessagePojo pojo = (MessagePojo) message1.getObject();
             // 该消费者用来让消息重发，从而进入死信队列
            session.recover();
            if (true) {
                return;
            }
            log.info(" {}-{} >>>> receiveQueueTest -2 接收到 msgBody:{} message:{}",appName,serverPort,pojo,message);
            sleep(1000);
            // 手动确认
            message.acknowledge();
        }
    }


    @JmsListener(destination = "delay-queue-test", containerFactory = "queueListener")
    public void receiveDelayQueueTest(ActiveMQMessage message, Session session) throws JMSException{
        log.info("receiveDelayQueueTest");
        String text = null;
        if (message instanceof TextMessage) {
            TextMessage textMessage = (TextMessage) message;
            text = textMessage.getText();
            log.info("delayQueue接收到消息：{}", text);
            ObjectMapper objectMapper = new ObjectMapper();
            sleep(2000);
            // 手动确认
            message.acknowledge();
        }
    }


    //监听死信队列
    @JmsListener(destination = "DLQ.queue-test")
    public void receiveDLQQueueTest(ActiveMQMessage message, Session session) throws JMSException {
        log.info("DLQ.queue-test");

        if(message instanceof  ObjectMessage){
            ObjectMessage message1 = (ObjectMessage) message;
            MessagePojo pojo = (MessagePojo) message1.getObject();
            log.info(" DLQ.queue-test  {}-{} >>>> 接收到消息  接收到 msgBody:{} message:{}",appName,serverPort,pojo,message);
            sleep(1000);
            // 手动确认
            message.acknowledge();
        }
    }



    private void sleep(long time) {
        try {
            Thread.sleep(time);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }


}
